package com.atuguigu.flink.Day01.Example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class FlatMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        DataStreamSource<String> stream = env.fromElements("white", "black", "gray");

        stream.flatMap(new MyFlatMap()).print();
        // 匿名函数的方式实现flatMap
        // 什么叫做匿名函数的类型？r -> r + 1; Integer -> Integer;
        // 必须指明函数的类型是`FlatMapFunction<String, String>)`

        stream.flatMap(
                (FlatMapFunction<String,String>)(s,out)->{
                    if (s.equals("while")) {
                        out.collect(s);
                    }

        }).print();



        env.execute();
    }

    public static class MyFlatMap implements FlatMapFunction<String,String>{

        @Override
        public void flatMap(String s, Collector<String> out) throws Exception {
            if(s.equals("white")){
                out.collect(s);
            }else if (s.equals("black")){
                out.collect(s);
                out.collect(s);
            }else {
                out.collect(s);
            }
        }
    }
}
